package com.steakliu.sun.client.netty.handler;

import com.alibaba.fastjson.JSONObject;
import com.steakliu.sun.buffer.api.DataBuffer;
import com.steakliu.sun.client.api.ClientConfig;
import com.steakliu.sun.client.netty.manager.ClientConnectionsManager;
import com.steakliu.sun.common.thread.ClientConcurrentThreadPool;
import com.steakliu.sun.common.thread.DataBufferThreadPool;
import com.steakliu.sun.constant.SerializerConstant;
import com.steakliu.sun.constant.SunConstant;
import com.steakliu.sun.protocol.*;
import com.steakliu.sun.protocol.context.RpcContext;
import com.steakliu.sun.protocol.future.SunFuture;
import com.steakliu.sun.protocol.generate.MsgHeaderGenerator;
import io.netty.channel.*;
import io.netty.handler.timeout.IdleStateEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 功能说明： client handler
 * <p>
 * Original @Author: steakliu-刘牌, 2023-01-08  12:35
 * <p>
 * Copyright (C)2020-2022  steakliu All rights reserved.
 */
public class SunClientHandler extends SimpleChannelInboundHandler<SunProtocol<SunResponse>> {

    private final Logger LOGGER = LoggerFactory.getLogger(SunClientHandler.class);

    private volatile Channel channel;

    private static final Map<Long, SunFuture> FUTURE = new ConcurrentHashMap<>();

    private final ClientConcurrentThreadPool clientConcurrentThreadPool;

    private final ClientConfig config;

    private DataBuffer dataBuffer;

    public SunClientHandler(ClientConcurrentThreadPool clientConcurrentThreadPool, ClientConfig config, DataBuffer requestBuffer){
        this.config = config;
        this.clientConcurrentThreadPool = clientConcurrentThreadPool;
        if (config.isEnableBuffer()) {
            this.dataBuffer = requestBuffer;
            DataBufferThreadPool.submit(() -> {
                this.handleRequestBuffer();
            });
        }
    }

    private void handleRequestBuffer(){
        while (true){
            SunProtocol protocol = dataBuffer.take();
            if (null != protocol){
                LOGGER.info("消费缓冲区来自【服务端】的响应：{}",JSONObject.toJSONString(protocol));
                this.handleResponse(protocol);
            }
        }
    }

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        this.channel = ctx.channel();
        super.channelRegistered(ctx);
    }

    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        ClientConnectionsManager.remove(channel);
        super.channelUnregistered(ctx);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        ClientConnectionsManager.remove(channel);
        super.channelInactive(ctx);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        LOGGER.info("client handler exception , ",cause);
        ClientConnectionsManager.remove(channel);
        super.exceptionCaught(ctx, cause);
    }

    /**
     * 事件触发
     *
     * @param ctx
     * @param event
     * @throws Exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object event) throws Exception {
        if (event instanceof IdleStateEvent) {
            SunProtocol<SunRequest> protocol = new SunProtocol<>();
            MsgHeader header = MsgHeaderGenerator.generateMsgHeader(SerializerConstant.JDK, MsgType.CLIENT_HEARTBEAT);
            SunRequest request = new SunRequest();
            request.setParameters(new Object[]{SunConstant.PING});
            protocol.setMsgHeader(header);
            protocol.setBody(request);
            //LOGGER.info("long time no received from server read event, send msg to server, msg content is : {}", JSONObject.toJSONString(protocol));
            channel.writeAndFlush(protocol);
        } else {
            super.userEventTriggered(ctx, event);
        }
    }

    @Override
    protected void channelRead0(ChannelHandlerContext context, SunProtocol<SunResponse> protocol) throws Exception {
        clientConcurrentThreadPool.submit(() -> {
            this.handle(protocol);
        });
    }

    private void handle(SunProtocol<SunResponse> protocol) {
        byte type = protocol.getMsgHeader().getType();
        MsgType msgType = MsgType.findType(type);
        if (null == msgType) {
            return;
        }
        switch (msgType) {
            case RESPONSE:
                this.handleResponseBuffer(protocol);
                break;
            case SERVER_RESPONSE_HEARTBEAT:
                this.handleServerResponseHeartBeat(protocol);
                break;
            default:
                throw new IllegalStateException("Unexpected value: " + msgType);
        }
    }

    /**
     * 处理服务端的心跳响应
     *
     * @param protocol
     */
    private void handleServerResponseHeartBeat(SunProtocol<SunResponse> protocol) {
        //LOGGER.info("heartbeat response received from the server, msg content is : {}", JSONObject.toJSONString(protocol));
    }

    private void handleResponseBuffer(SunProtocol<SunResponse> protocol){
        if (config.isEnableBuffer()){
            dataBuffer.push(protocol);
        }else {
            this.handleResponse(protocol);
        }
    }

    /**
     * 处理服务端的响应
     * @param protocol
     */
    private void handleResponse(SunProtocol<SunResponse> protocol) {
        //获取服务端响应数据后，加入Future中
        long requestId = protocol.getMsgHeader().getRequestId();
        SunFuture sunFuture = FUTURE.remove(requestId);
        if (null != sunFuture) {
            sunFuture.done(protocol);
        }
    }


    public SunFuture sendRequest(SunProtocol<SunRequest> protocol, ClientConfig config) {
        return clientConcurrentThreadPool.submit(() ->
                config.isAsync() ? this.handleAsync(protocol) : config.isOneway() ? this.handleOneway(protocol) : this.handleSync(protocol)
        );
    }

    /**
     * 单向调用
     *
     * @param protocol
     * @return
     */
    private SunFuture handleOneway(SunProtocol<SunRequest> protocol) {
        this.channel.writeAndFlush(protocol);
        return null;
    }

    /**
     * 异步调用
     * @param protocol
     * @return
     */
    private SunFuture handleAsync(SunProtocol<SunRequest> protocol) {
        SunFuture sunFuture = SunFuture.getInstance();
        FUTURE.put(protocol.getMsgHeader().getRequestId(), sunFuture);
        this.channel.writeAndFlush(protocol);
        return sunFuture;
    }

    /**
     * 同步调用
     *
     * @param protocol
     * @return
     */
    private SunFuture handleSync(SunProtocol<SunRequest> protocol) {
        SunFuture sunFuture = SunFuture.getInstance();
        FUTURE.put(protocol.getMsgHeader().getRequestId(), sunFuture);
        this.channel.writeAndFlush(protocol);
        return sunFuture;
    }


}
